package com.cti.commons.mqtt.handle;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.alibaba.fastjson.JSON;
import com.cti.commons.mqtt.domain.MessageDTO;
import com.cti.commons.mqtt.repository.MqttSendMessageLog;
import com.cti.commons.mqtt.repository.MqttSendMessageRepository;

import cn.hutool.core.convert.Convert;
import lombok.extern.slf4j.Slf4j;

/**
 * @author et.wang
 * @description: 消息发送抽象类
 * @date 2022/11/15 10:15
 */
@Slf4j
public abstract class AbstractMessageSender {

    public static final int SIZE = 1024;

    @Resource(name = "mqttSenderExecutor")
    protected ThreadPoolTaskExecutor executor;

    @Autowired
    private MqttMessageGateway messageGateway;

    @Autowired
    private MqttSendMessageRepository repository;

    public void send(String topic, int qos, MessageDTO message) {
        MqttSendMessageLog mqttSendMessageLog = new MqttSendMessageLog(message.getId(), topic,
            Convert.toStr(message.getPayload()),
            message.getBusinessId(),
            qos);
        try {
            repository.save(mqttSendMessageLog);
        } catch (DuplicateKeyException e) {
            throw new RuntimeException("messageId重复:" + message.getId());
        }
        /**
         * 异步发送消息和更新消息日志
         */
        //this.executor.execute(() -> {
        try {
            log.info("发送消息: topic:{},message:{}", topic, JSON.toJSONString(message));
            messageGateway.send(topic, qos, message.getPayload());
            mqttSendMessageLog.success();
        } catch (Exception e) {
            log.error("发送失败:", e);
            mqttSendMessageLog.fail();
            String errorMessageString = e.getMessage();
            mqttSendMessageLog.setException(errorMessageString.length() < SIZE ? errorMessageString :
                errorMessageString.substring(0, SIZE));
        }
        repository.save(mqttSendMessageLog);
        //});

    }

}
